package com.etc

/**
  * @Title:
  * @ProjectName
  * @Description:
  * @author kalista
  */
//import com.etc.MyZkSerializer
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 针对的kafka集群时0.10
  */
object KafkaDirect2 {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("DirectKafkaMeterData")
    val ssc = new StreamingContext(conf, Seconds(30))//流数据分批处理时间间隔
    //kafka节点
    val BROKER_LIST = "master:9092,slave1:9092,slave2:9092"
    val ZK_SERVERS = "master:2181,slave1:2181,slave2:2181"
    val GROUP_ID = "test-consumer-group" //消费者组
    val topics = Array("my-topic") //待消费topic

    /*
    参数说明
    AUTO_OFFSET_RESET_CONFIG
        earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
        latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
        none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
     */
    val kafkaParams = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> BROKER_LIST,
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> GROUP_ID,
      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "latest",
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean)
    )

    //采用zookeeper手动维护偏移量
    val zkManager = new KafkaOffsetZKManager(ZK_SERVERS)

    /**
      * key    topic+partition    偏移量
      */
    val fromOffsets = zkManager.getFromOffset(topics,GROUP_ID)

    //创建数据流
    var stream:InputDStream[ConsumerRecord[String, String]] = null

    if (fromOffsets.size > 0){
      stream = KafkaUtils.createDirectStream[String, String](
        ssc,
        PreferConsistent,
        Subscribe[String, String](topics, kafkaParams, fromOffsets)
      )
    }else{
      stream = KafkaUtils.createDirectStream[String, String](
        ssc,
        PreferConsistent,
        Subscribe[String, String](topics, kafkaParams)
      )
      println("第一次消费 Topic:" + topics)
    }

    //处理流数据
    stream.foreachRDD { rdd =>
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      val rs = rdd.map(record => (record.offset(), record.partition(), record.value())).collect()
      for(item <- rs) println(item)
      // 处理数据存储到HDFS或Hbase等
      // 存储代码（略）
      // 处理完数据保存/更新偏移量
      zkManager.storeOffsets(offsetRanges,GROUP_ID)
    }

    ssc.start()
    ssc.awaitTermination()
  }

}
